package worker;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.StringReader;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.Arrays;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.PriorityQueue;

import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Marshaller;
import javax.xml.bind.Unmarshaller;
import javax.xml.transform.stream.StreamSource;

import common.FileObject;
import common.JobObject;
import common.JobRequestObject;
import common.JobResultObject;
import common.MasterObject;
import common.WorkerObject;

public class WorkerService {

	private InetAddress registryIP;
	private int registryUDPPort;
	private int masterTCPPort = 5002;
	private WorkerObject worker;
	private PriorityQueue<JobObject> jobQueue;
	private List<JobObject> completedJobList;
	private MasterObject master;

	public WorkerService (InetAddress registryIP, int registryUDPPort, int masterTCPPort, WorkerObject worker, PriorityQueue<JobObject> jobQueue, List<JobObject> completedJobList) {
		this.registryIP = registryIP;
		this.registryUDPPort = registryUDPPort;
		this.worker = worker;
		this.jobQueue = jobQueue;
		this.completedJobList = completedJobList;
	}

	public void register(){

		System.out.println("SYSINFO - Worker registration service start");

		DatagramSocket aSocket = null;

		try {
			//			String regXML = createRegXML();
			JAXBContext context = JAXBContext.newInstance(RegisterRequestObject.class);

			Marshaller m = context.createMarshaller();
			m.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);

			RegisterRequestObject registerRequest = new RegisterRequestObject();
			registerRequest.setFrom("Worker");
			registerRequest.setType("Register");
			registerRequest.setTimestamp(getTime());
			int load = 0;
			Iterator<JobObject> iter = jobQueue.iterator();
			while (iter.hasNext()) {
				load += iter.next().getTime();
			}
			worker.setLoad(load);
			registerRequest.setWorker(worker);


			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			m.marshal(registerRequest, baos);
			String registerXML = baos.toString();
			System.out.println("DEBUG - XML Request:");
			System.out.println(registerXML);

			aSocket = new DatagramSocket();
			byte[] buf = registerXML.getBytes();

			DatagramPacket request = new DatagramPacket(buf, buf.length, registryIP, registryUDPPort);
			aSocket.send(request);
			System.out.println("SYSINFO - Send registration request to RD: "+registryIP.getHostAddress()+":"+registryUDPPort);
		} catch (SocketException e){System.out.println("Socket:" + e.getMessage());
		} catch (IOException e){System.out.println("IO:" + e.getMessage());
		} catch (Exception e) {e.printStackTrace();
		} finally {
			if(aSocket != null) {
				aSocket.close();
			}
		}
	}

	public void ReceiveJobs(PriorityQueue<JobObject> jobQueue, WorkerService workerService) {

		System.out.println("SYSINFO - Worker ReceiveJob service start.");

		try{
			ServerSocket listenSocket = new ServerSocket(worker.getPort());

			while (true) {
				Socket masterSocket = listenSocket.accept();
				System.out.println("SYSINFO - Accept Master TCP connection for receving a job.");
				ReceiveAJob(masterSocket);
			}
		}catch(IOException e){System.out.println("Listen:"+e.getMessage());}
	}

	public void executeFile(JobObject job) {

		System.out.println("SYSINFO - Worker executeFile service start");

		String lineStr = null;
		String outputfile = job.getExecuteFileName() + ".out" + job.getClientID();
		String errorfile = job.getExecuteFileName() + ".err" + job.getClientID();

		try {
			String cmd = "java -classpath "+job.getExecuteFilePath()+" "+job.getExecuteFileName()+ " " + job.getParameters();
			System.out.println("DEBUG - cmd: "+cmd);
			Process p = Runtime.getRuntime().exec(cmd);
			//			Process p = Runtime.getRuntime().exec("java worker.HelloWorld Melbourne");

			BufferedInputStream in = new BufferedInputStream(p.getInputStream());
			BufferedInputStream err = new BufferedInputStream(p.getErrorStream());
			BufferedReader inBr = new BufferedReader(new InputStreamReader(in));
			BufferedReader errBr = new BufferedReader(new InputStreamReader(err));

			FileOutputStream fos = new FileOutputStream(job.getExecuteFilePath()+"/"+outputfile);
			// read the output from the command
			while ((lineStr = inBr.readLine()) != null) {
				System.out.println(lineStr);
				fos.write((lineStr+"\n").getBytes());
			}
			fos.close();

			fos = new FileOutputStream(job.getExecuteFilePath()+"/"+errorfile);
			// read any errors from the attempted command
			while ((lineStr = errBr.readLine()) != null) {
				System.out.println(lineStr);
				fos.write((lineStr+"\n").getBytes());
			}
			fos.close();

			try {
				if (p.waitFor()!=0) {
					if(p.exitValue()==1) System.err.println("fail!");
				}
			}catch (InterruptedException e){
				e.printStackTrace();
			}

			System.out.println("DEBUG: Job is Completed.");
			completedJobList.add(job);

		}
		catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void ReceiveAJob(Socket masterSocket) {

		DataInputStream dataIn = null;
		JobObject job = null;
		try {
			InputStream in = masterSocket.getInputStream();
			dataIn = new DataInputStream(in);

			/********** JobPackage Information Transfer **********/
			String jobRequestXML = dataIn.readUTF();
			System.out.println("SYSINFO - Received a job request.");
			System.out.println("DEBUG - Request XML: "+jobRequestXML);

			JAXBContext context = JAXBContext.newInstance(JobRequestObject.class);
			Unmarshaller u = context.createUnmarshaller();
			StringBuffer xmlStr = new StringBuffer(jobRequestXML);
			JobRequestObject JobRequest = (JobRequestObject) u.unmarshal(new StreamSource(new StringReader(xmlStr.toString())));
			if (JobRequest.getFrom().equals("Master") && JobRequest.getType().equals("SendJob")) {
				job = JobRequest.getJobObject();

				Long fileSize = dataIn.readLong();

				System.out.println("DEBUG - fileName = " + job.getExecuteFileName());
				System.out.println("DEBUG - parameters = "	+ job.getParameters());
				System.out.println("DEBUG - time = " + job.getTime());
				System.out.println("DEBUG - time = " + job.getExtraTime());
				System.out.println("DEBUG - jobID = " + job.getJobID());
				System.out.println("DEBUG - priority = " + job.getPriority());
				System.out.println("DEBUG - priority = " + job.getStatus());
				System.out.println("DEBUG - priority = " + job.getPriority());
				System.out.println("DEBUG - clientID = " + job.getClientID());
				System.out.println("DEBUG - fileSize = " + fileSize);

				String filePath = "workerJobs"+"/"+job.getClientID(); // default jobs folder (but can be assign to another one)
				System.out.println(filePath);
				File file = new File(filePath);
				if (file.exists()==false) {
					file.mkdirs();
				}
				job.setExecuteFilePath(filePath);
				jobQueue.add(job);
			} else {
				System.out.println("SYSINFO - The request is not a JobRequest. No action will be performed.");
			}
			
			/********** File Transfer **********/
			// NTD: handle multiple files transfer
			String savePath = job.getExecuteFilePath();
			int bufferSize = 8192;
			byte[] buf = new byte[bufferSize];
			savePath += "/"+job.getExecuteFileName()+".class";
			DataOutputStream fileOut = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(savePath)));

			int length = 0;
			while((length = dataIn.read(buf))!=-1) {
				fileOut.write(buf, 0, length);
			}
			System.out.println("File saved at " + savePath + "");
			fileOut.close();

			masterSocket.close();
			System.out.println("SYSINFO - Master TCP connection is close.");
		} catch(EOFException e) {System.out.println("EOF:"+e.getMessage());
		} catch(IOException e) {System.out.println("IO:"+e.getMessage());
		} catch (Exception e) {e.printStackTrace();
		} finally{
			try{
				if(masterSocket != null) masterSocket.close();
			}
			catch(IOException e){/*close failed*/}
		}
	}

	public void sendBackResult(JobObject job) {

		Socket socket = null;
		try {
			// socket = (SSLSocket) sslsocketfactory.createSocket(workerIP,
			// workerTCPPort);registryIP
			socket = new Socket(registryIP, masterTCPPort);
			OutputStream out = socket.getOutputStream();
			DataOutputStream dataOut = new DataOutputStream(out);

			/********** JobResult Information Transfer **********/	
			String filePath = job.getExecuteFilePath();
			File file = new File(filePath);
			File fileList[] = file.listFiles();
			List<File> fileLists = Arrays.asList(fileList);
			String jobResult = createJobResultXML(job, fileLists);
			System.out.println("DEBUG - XML Result:");
			System.out.println(jobResult);
			
			dataOut.writeUTF(jobResult);
			System.out.println("SYSINFO - Send a Job Information.");
			
			/********** File Transfer **********/
			// NTD: file filter
			Iterator<File> iter = fileLists.iterator();
			while(iter.hasNext()){
				File resultFile = iter.next();
				if(resultFile.isFile()){
					System.out.println("DEBUG - File name:" +  resultFile.getName());

					// send the file content
					DataInputStream fileIn = new DataInputStream(new BufferedInputStream(new FileInputStream(resultFile)));
					int bufferSize = 8192;
					byte[] buf = new byte[bufferSize];
					int length = 0;
					while((length = fileIn.read(buf))!=-1) {
						dataOut.write(buf, 0, length);
					}
					dataOut.flush();
				}
				else{
					System.out.println("SYSINFO - "+file.getName()+" is a directory.");
				}
				System.out.println("DEBUG - Send a output file.");
			}
			System.out.println("SYSINFO - Send a job Result.");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			if(socket!=null) {
				try {
					socket.close();
					System.out.println("SYSINFO - Worker to Master TCP connection is close.");
				} catch(IOException e){/*close failed*/}
			}
		}

	}

	private String createJobResultXML(JobObject job, List<File> fileLists) {
		
		String jobRequestXML = null;
		try {
			JAXBContext context = JAXBContext.newInstance(JobResultObject.class);
			Marshaller m = context.createMarshaller();
		    m.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);

		    JobResultObject jobResult = new JobResultObject();
		    
		    jobResult.setFrom("Worker");
		    jobResult.setType("SendJobResult");
		    jobResult.setTimestamp(getTime());
		    jobResult.setJob(job);
		    jobResult.setNumberOfOutputFiles(fileLists.size());
		    jobResult.setWorker(worker);
		    
		    if (!fileLists.isEmpty()) {
		    	Iterator<File> iter = fileLists.iterator();
		    	while (iter.hasNext()) {
					File file = iter.next();
					if (file.isFile()) {
						FileObject outputFile = new FileObject();
						outputFile.setFileName(file.getName());
						outputFile.setFileSize(file.length());
						jobResult.addFile(outputFile);
					}
				}
		    }
		    
		    ByteArrayOutputStream baos = new ByteArrayOutputStream();
		    m.marshal(jobResult, baos);
		    jobRequestXML = baos.toString();
		    
		} catch (JAXBException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
		return jobRequestXML;
	}
	
	public Long getTime() {
		Date date = new Date();
		return date.getTime();
	}

}
